////////////////////////////////////////////////////////////////////////////
//
//  CryEngine Source File.
//  Copyright (C), Crytek.
// -------------------------------------------------------------------------
//  File name:   StreamIOThread.h
//  Created:     22/07/2010 by Timur.
//  Description: Streaming Thread for IO
// -------------------------------------------------------------------------
//  History:
//
////////////////////////////////////////////////////////////////////////////

#ifndef __StreamingIOThread_h__
#define __StreamingIOThread_h__
#pragma once

#include <IStreamEngine.h>
#include "StreamAsyncFileRequest.h"

class CStreamEngine;

//////////////////////////////////////////////////////////////////////////
// Thread that performs IO operations.
//////////////////////////////////////////////////////////////////////////
class CStreamingIOThread : public CrySimpleThread<CStreamingIOThread>, public CMultiThreadRefCount
{
public:
	CStreamingIOThread( CStreamEngine *pStreamEngine,EStreamSourceMediaType mediaType,const char *name, int32 prio=THREAD_PRIORITY_NORMAL );
	~CStreamingIOThread();

	void CancelAll();
	void AbortAll(bool bAbort);

	void BeginReset();
	void EndReset();

	void AddRequest( CAsyncIOFileRequest* pRequest,bool bStartImmidietly );
	int GetRequestCount() const { return m_fileRequestQueue.size(); };
	void SortRequests();
	void NeedSorting();
	void SignalStartWork(bool bForce);

	void Pause(bool bPause);

	void RegisterFallbackIOThread(EStreamSourceMediaType mediaType, CStreamingIOThread* pIOThread);

	//////////////////////////////////////////////////////////////////////////
	// CrySimpleThread
	//////////////////////////////////////////////////////////////////////////
	virtual void Run();
	virtual void Cancel();
	//////////////////////////////////////////////////////////////////////////

protected:

	void ProcessNewRequests();
	void ProcessReset();

public:

	struct SStats
	{
		SStats() : m_nTotalReadBytes(0), m_nCurrentReadBandwith(0),
			m_nReadBytesInLastSecond(0), m_fReadingDuringLastSecond(.0f),
			m_nTempBytesRead(0), m_nActualReadBandwith(0), m_nTempReadOffset(0),
			m_nTotalReadOffset(0), m_nReadOffsetInLastSecond(0), m_nTempRequestCount(0),
			m_nTotalRequestCount(0), m_nRequestCountInLastSecond(0)
		{}

		void Update(const CTimeValue& deltaT);

		void Reset()
		{
			m_nTotalReadBytes = 0;
			m_nTotalReadOffset = 0;
			m_nTotalRequestCount = 0;
			m_TotalReadTime.SetValue(0);
		}

		float m_fReadingDuringLastSecond;
		CTimeValue m_TotalReadTime;
		uint64 m_nTotalReadBytes;
		uint64 m_nTotalReadOffset;
		uint32 m_nTotalRequestCount;
		uint32 m_nCurrentReadBandwith;			// Read bandwidth over one second
		uint32 m_nActualReadBandwith;				// Actual read bandwidth extrapolated over one second
		uint32 m_nReadBytesInLastSecond;
		uint32 m_nRequestCountInLastSecond;
		uint64 m_nReadOffsetInLastSecond;

		uint32 m_nTempRequestCount;
		uint64 m_nTempBytesRead;
		uint64 m_nTempReadOffset;
		CTimeValue m_TempReadTime;
	};
	
	SStats m_InMemoryStats;
	SStats m_NotInMemoryStats;

	int64 m_nLastReadDiskOffset;
	int m_nStreamingCPU;

private:
	CStreamEngine *m_pStreamEngine;
	std::vector<CAsyncIOFileRequest*> m_fileRequestQueue;
	std::vector<CAsyncIOFileRequest*> m_temporaryArray;
	CryMT::vector<CAsyncIOFileRequest*> m_newFileRequests;

	EStreamSourceMediaType m_eMediaType;

	typedef std::pair<CStreamingIOThread*,EStreamSourceMediaType> TFallbackIOPair;
	typedef std::vector<TFallbackIOPair> TFallbackIOVec;
	typedef TFallbackIOVec::iterator TFallbackIOVecConstIt;
	TFallbackIOVec m_FallbackIOThreads;

	volatile bool m_bCancelThreadRequest;
	volatile bool m_bNeedSorting;
	volatile bool m_bNewRequests;
	volatile bool m_bPaused;
	volatile bool m_bNeedReset;
	volatile bool m_bAbortReads;

	volatile int m_iUrgentRequests;

	CryEvent m_awakeEvent;
	CryEvent m_resetDoneEvent;
	string m_name;
	uint32 m_nReadCounter;
};

//////////////////////////////////////////////////////////////////////////
// Thread that performs IO operations.
//////////////////////////////////////////////////////////////////////////
class CStreamingWorkerThread : public CrySimpleThread<CStreamingIOThread>, public CMultiThreadRefCount
{
public:
	enum EWorkerType
	{
		eWorkerDecompress,
		eWorkerAsyncCallback,
	};
	CStreamingWorkerThread( CStreamEngine *pStreamEngine,const char *name,EWorkerType type );
	~CStreamingWorkerThread();

	void BeginReset();
	void EndReset();

	void CancelAll();

	void AddRequest( CAsyncIOFileRequest* pRequest );
	int GetRequestCount() const { return m_fileRequestQueue.size(); };

	//////////////////////////////////////////////////////////////////////////
	// CrySimpleThread
	//////////////////////////////////////////////////////////////////////////
	virtual void Run();
	virtual void Cancel();
	//////////////////////////////////////////////////////////////////////////

private:
	EWorkerType m_type;
	CStreamEngine *m_pStreamEngine;
	CryMT::vector<CAsyncIOFileRequest_AutoPtr> m_fileRequestQueue;

	volatile bool m_bCancelThreadRequest;
	volatile bool m_bNeedsReset;
	
	CryEvent m_awakeEvent;
	CryEvent m_resetDoneEvent;
	string m_name;
};

#endif //__StreamingIOThread_h__
